package com.flink.sink;

import com.flink.source.Data;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.Timestamp;
import java.util.logging.Logger;

/**
 * @author lzl
 * @create 2024-01-22 15:30
 * @name MySqlToPojoSink
 */
public class MySqlToPojoSink extends RichSinkFunction<Data> {

    private static final Logger log = Logger.getLogger(MySqlToPojoSink.class.getName());
    private static final long serialVersionUID = 1L;
    private Connection connection = null;
    private PreparedStatement ps = null;
    private String tableName = "test";

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        log.info("获取数据库连接");

        // 通过Druid获取数据库连接，准备写入数据库
        connection = DbUtils.getConnection();

        // 插入数据库的语句   因为我们封装的pojo的类型为PojoType<com.lzl.flink.Web, fields = [area: String, createDate: Date, ip: String, operate: String, uuid: String, web: String]>
        String insertQuery = "INSERT INTO " +   tableName + "(time,ip,uid,area,web,operate) VALUES (?,?,?,?,?,?)" ;

        // 执行插入语句
        ps = connection.prepareStatement(insertQuery);
    }

    // 重新关闭方法。   关闭并释放资源
    @Override
    public void close() throws Exception {
        super.close();
        if(connection != null) {
            connection.close();
        }
        if (ps != null ) {
            ps.close();
        }
    }

    // 重写invoke方法
    @Override
    public void invoke(Data value,Context context) throws Exception {
        //组装数据，执行插入操作
        ps.setTimestamp(1, new Timestamp(value.getCreateDate().getTime()));
        ps.setString(2,value.getIp());
        ps.setString(3, value.getUuid());
        ps.setString(4, value.getArea());
        ps.setString(5, value.getWeb());
        ps.setString(6, value.getOperate());
        ps.addBatch();

        // 一次性写入
        int[] count = ps.executeBatch();
        System.out.println("成功写入MySQL数量：" + count.length);
    }
}
